package com.example.mq.mqserver.datacenter;



import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

public class MessageFileManager {


    public void init() {
        // 暂时不需要做啥额外的初始化工作, 以备后续扩展
    }

    static public class Stat{

         public int totalCount;
         public int validCount;
     }

    // 预定消息文件所在的目录和文件名
    // 这个方法, 用来获取到指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 这个方法用来获取该队列的消息数据文件路径
    // 注意, 二进制文件, 使用 txt 作为后缀, 不太合适. txt 一般表示文本. 此处咱们也就不改.
    // .bin / .dat
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }
    // 这个方法用来获取该队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName) {
         Stat stat = new Stat();
         try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
             Scanner scanner = new Scanner(inputStream);
             stat.totalCount = scanner.nextInt();
             stat.validCount = scanner.nextInt();
             return stat;
         }  catch (IOException e) {
             e.printStackTrace();
         }
         return null;
    }
    private void writeStat(String queueName,Stat stat) {
         try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
               PrintWriter printWriter = new PrintWriter(outputStream);
               printWriter.write(stat.totalCount + "\t" + stat.validCount);
               printWriter.flush();
         } catch (IOException e) {
             e.printStackTrace();
         }
    }
    //创建消息对应的目录和文件
    public void createQueueFiles(String queueName) throws IOException {
    // 1. 先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 不存在, 就创建这个目录
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给消息统计文件, 设定初始值. 0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }
    // 删除队列的目录和文件.
    // 队列也是可以被删除的. 当队列删除之后, 对应的消息文件啥的, 自然也要随之删除.
    public void destroyQueueFiles(String queueName) throws IOException {
         File queueDataFile = new File(getQueueDataPath(queueName));
         boolean ok1 = queueDataFile.delete();
         File queueStatFile = new File(getQueueStatPath(queueName));
         boolean ok2 = queueStatFile.delete();
         File baseDir = new File(getQueueDir(queueName));
         boolean ok3 = baseDir.delete();
         if (!ok1 || !ok2 || !ok3) {
             // 有任意一个删除失败, 都算整体删除失败.
             throw new IOException("删除队列目录和文件失败! baseDir=" + baseDir.getAbsolutePath());
         }
    }
    // 检查队列的目录和文件是否存在.
    // 比如后续有生产者给 broker server 生产消息了, 这个消息就可能需要记录到文件上(取决于消息是否要持久化)
    public boolean checkFileExists(String queueName) {
         File queueDataFile = new File(getQueueDataPath(queueName));
         boolean ok1 = queueDataFile.exists();
         if (!ok1) {
             return false;
         }
         File queueStatFile = new File(getQueueStatPath(queueName));
         boolean ok2 = queueStatFile.exists();
         if (!ok2) {
             return false;
         }
         return true;
    }

    // 这个方法用来把一个新的消息, 放到队列对应的文件中.
    // queue 表示要把消息写入的队列. message 则是要写的消息.
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
         //1.判断文件是否存在
         if (!checkFileExists(queue.getName())) {
             throw  new MqException("[MessageFileManager]对应的队列文件不存在! queueName = " + queue.getName());
         }
         //2.将message对象序列化，转成二进制数组
         byte[] messageBinary = BinaryTool.toBytes(message);
         synchronized (queue){
             //3.获取当前队列的文件对象
             // 把新的 Message 数据, 写入到队列数据文件的末尾. 此时 Message 对象的 offsetBeg , 就是当前文件长度 + 4
             // offsetEnd 就是当前文件长度 + 4 + message 自身长度.

             File queueDataFile = new File(getQueueDataPath(queue.getName()));
             message.setOffsetBeg(queueDataFile.length() + 4);
             message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
             //4.写入消息到消息文件，注意这里是追加在文件末尾
             try(OutputStream outputStream = new FileOutputStream(queueDataFile,true)) {
                 try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                     //写入消息长度，占四个字节
                     dataOutputStream.writeInt(messageBinary.length);
                     //在写入消息本体
                     dataOutputStream.write(messageBinary);
                 }
             }
             //5.更新消息统计文件
             Stat stat = new Stat();
             stat.totalCount += 1;
             stat.validCount += 1;
             writeStat(queue.getName(),stat);
         }
    }
    // 这个是删除消息的方法.
    // 这里的删除是逻辑删除, 也就是把硬盘上存储的这个数据里面的那个 isValid 属性, 设置成 0
    // 1. 先把文件中的这一段数据, 读出来, 还原回 Message 对象;
    // 2. 把 isValid 改成 0;
    // 3. 把上述数据重新写回到文件.
    // 此处这个参数中的 message 对象, 必须得包含有效的 offsetBeg 和 offsetEnd
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
         synchronized (queue) {
             try( RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw");) {
                 //1.先从文件中读取messag数据
                 byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                 randomAccessFile.seek(message.getOffsetBeg());
                 randomAccessFile.read(bufferSrc);
                 //2.吧当前读取到的数据转成message对象
                 Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                 //3.将有效位变为0
                 diskMessage.setIsValid((byte) 0x0);
                 //4.将message写回文件中
                 byte[] bufferMessage = BinaryTool.toBytes(diskMessage);
                 randomAccessFile.seek(message.getOffsetBeg());
                 randomAccessFile.write(bufferMessage);

                 //5.更新消息文件
                 Stat stat = new Stat();
                 if (stat.validCount > 0) {
                     stat.validCount -= 1;
                 }
                 writeStat(queue.getName(),stat);
             }
         }

    }
    // 使用这个方法, 从文件中, 读取出所有的消息内容, 加载到内存中(具体来说是放到一个链表里)
    // 这个方法, 准备在程序启动的时候, 进行调用.
    // 这里使用一个 LinkedList, 主要目的是为了后续进行头删操作.
    // 这个方法的参数, 只是一个 queueName 而不是 MSGQueue 对象. 因为这个方法不需要加锁, 只使用 queueName 就够了.
    // 由于该方法是在程序启动时调用, 此时服务器还不能处理请求呢~~ 不涉及多线程操作文件.
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                //这个变量记录当前文件遍历光标
                int currentOffset = 0;
                while (true) {
                    //1.读取当前消息的长度
                    int messageSize = dataInputStream.readInt();
                    byte[] buffer = new byte[messageSize];
                    //实际读取的长度
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        throw new MqException("[MessageFileManager] 文件读取错误 queueName = " + queueName);
                    }
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    if (message.getIsValid() != 0x1) {
                        currentOffset += (4 + messageSize);
                      //无效数据，直接跳过
                        continue;
                    }
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            }catch (EOFException e) {
                // 这个 catch 并非真是处理 "异常", 而是处理 "正常" 的业务逻辑. 文件读到末尾, 会被 readInt 抛出该异常.
                // 这个 catch 语句中也不需要做啥特殊的事情
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
             return messages;
    }
    //检查当前文件中的消息是否需要进行GC
    public boolean checkGC(String queueName) {
       Stat stat = readStat(queueName);
       if (stat.totalCount > 2000 && (double)stat.validCount / (double) stat.totalCount < 0.5) {
           return true;
       }
       return false;
    }

    private String getQueueDataNewPath(String queueName) {
         return getQueueDir(queueName )+ "queue_data_new.txt";
    }

    // 通过这个方法, 真正执行消息数据文件的垃圾回收操作.
    // 使用复制算法来完成.
    // 创建一个新的文件, 名字就是 queue_data_new.txt
    // 把之前消息数据文件中的有效消息都读出来, 写到新的文件中.
    // 删除旧的文件, 再把新的文件改名回 queue_data.txt
    // 同时要记得更新消息统计文件.
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 进行 gc 的时候, 是针对消息数据文件进行大洗牌. 在这个过程中, 其他线程不能针对该队列的消息文件做任何修改.
         synchronized (queue) {
             // 由于 gc 操作可能比较耗时, 此处统计一下执行消耗的时间.
             long gcBeg = System.currentTimeMillis();

             File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
             if (queueDataNewFile.exists()) {
                 throw new MqException("[MessageFileManager] gc时该队列的queue_data_new 已经存在 queueName = " + queue.getName());
             }
            boolean ok =  queueDataNewFile.createNewFile();
             if (!ok) {
                 throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
             }
             //从旧文件中读取有效消息
             LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

             //3.把有效消息写入新文件
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }

             // 4. 删除旧的数据文件, 并且把新的文件进行重命名
             File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
             ok = queueDataOldFile.delete();
             if (!ok) {
                 throw new MqException("[MessageFileManager] 删除旧的数据文件失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
             }
             // 把 queue_data_new.txt => queue_data.txt
             ok = queueDataNewFile.renameTo(queueDataOldFile);
             if (!ok) {
                 throw new MqException("[MessageFileManager] 文件重命名失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath()
                         + ", queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
             }

             // 5. 更新统计文件
             Stat stat = readStat(queue.getName());
             stat.totalCount = messages.size();
             stat.validCount = messages.size();
             writeStat(queue.getName(), stat);
             long gcEnd = System.currentTimeMillis();
             System.out.println("[MessageFileManager] gc 执行完毕! queueName=" + queue.getName() + ", time="
                     + (gcEnd - gcBeg) + "ms");
         }
    }
}
